package com.atguigu.edu.realtime.common.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.util.HBaseUtil;
import com.atguigu.edu.realtime.common.util.RedisUtil;
import io.lettuce.core.api.StatefulRedisConnection;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.hadoop.hbase.client.AsyncConnection;

import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> implements DimFunction<T> {
    private AsyncConnection asyncHbaseConn;
    private StatefulRedisConnection<String, String> asyncRedisConn;

    @Override
    public void open(Configuration parameters) throws Exception {
        asyncHbaseConn = HBaseUtil.getAsyncConnection();
        asyncRedisConn = RedisUtil.getAsyncRedisConnection();
    }

    @Override
    public void close() throws Exception {
        HBaseUtil.closeAsyncConnection(asyncHbaseConn);
        RedisUtil.closeAsyncRedisConnection(asyncRedisConn);
    }

    @Override
    public void asyncInvoke(T obj, ResultFuture<T> resultFuture) throws Exception {
//创建异步编排对象  执行线程任务，有返回结果，这个线程任务的返回结果将作为下一个线程任务的入参
        CompletableFuture.supplyAsync(
                new Supplier<JSONObject>() {
                    @Override
                    public JSONObject get() {
                        //从redis获取维度数据
                        JSONObject dimJsonObj = RedisUtil.readDimAsync(asyncRedisConn, getTableName(), getRowKey(obj));
                        return dimJsonObj;
                    }
                }
                //有入参，有返回值
        ).thenApplyAsync(
                new Function<JSONObject, JSONObject>() {
                    @Override
                    public JSONObject apply(JSONObject dimJsonObj) {
                        if (dimJsonObj != null) {
                            //如果从redis中找到了要关联的维度数据----缓存命中
                            System.out.println("~~~从Redis中获取" + getTableName() + "的" + getRowKey(obj) + "数据~~~");
                        } else {
                            //如果从redis中没有找到要关联的维度数据，发送请求到hbase中查询维度数据
                            dimJsonObj = HBaseUtil.getRowAsync(asyncHbaseConn, Constant.HBASE_NAMESPACE, getTableName(), getRowKey(obj));
                            //将查询出来的维度数据 放到redis中缓存起来，方便下次查询使用
                            if (dimJsonObj != null) {
                                System.out.println("~~~从HBase中获取" + getTableName() + "的" + getRowKey(obj) + "数据~~~");
                                RedisUtil.writeDimAsync(asyncRedisConn, getTableName(), getRowKey(obj), dimJsonObj);
                            } else {
                                //从HBase中也没有找到要关联的维度
                                System.out.println("~~~从HBase中没有找到" + getTableName() + "的" + getRowKey(obj) + "数据~~~");
                            }

                        }
                        return dimJsonObj;
                    }
                }
        ).thenAcceptAsync(
                //有入参，无返回值
                new Consumer<JSONObject>() {
                    @Override
                    public void accept(JSONObject dimJsonObj) {
                        if (dimJsonObj != null) {
                            //将维度属性补充到流中对象上
                            addDims(obj, dimJsonObj);
                        }
                        //将关联后的数据传递到下游
                        resultFuture.complete(Collections.singleton(obj));
                    }
                }
        );
    }

    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        throw new RuntimeException("如果异步关联维度超时，可以做如下检查：\n" +
                "1.是否将相关联组件都启动了:zk、kafka、hdfs、hbase、redis \n" +
                "2.查看Redis中bind的配置，是否注掉或者0.0.0.0\n" +
                "3.查看HBase维度表中是否存在维度数据\n" +
                "4.找昌平彦祖\n");
    }
}